package fr.jade.fraclite.orb.jgroups;

import java.util.HashMap;
import java.util.Map;
import java.util.Vector;
import java.util.logging.Logger;

import org.jgroups.Channel;
import org.jgroups.MembershipListener;
import org.jgroups.Message;
import org.jgroups.MessageListener;
import org.jgroups.blocks.MethodCall;
import org.jgroups.blocks.RspFilter;
import org.jgroups.util.RspList;

public class RpcDispatcher extends org.jgroups.blocks.RpcDispatcher {

  protected Logger log = Logger.getLogger(getClass().getName());
  protected Map<String, Call> calls;
  
  public RpcDispatcher(Channel channel, MessageListener l, MembershipListener l2, Object server_obj) {
    super(channel, l, l2, server_obj);
    calls = new HashMap<String, Call>();
  }

  public RspList callRemoteMethods(String callId, Vector dests, String method_name, Object[] args, Class[] types, int mode, long timeout) {
    return callRemoteMethods(callId, dests, method_name, args, types, mode, timeout, false);
  }

  public RspList callRemoteMethods(String callId, Vector dests, String method_name, Object[] args, Class[] types, int mode, long timeout, boolean use_anycasting) {
    return callRemoteMethods(callId, dests, method_name, args, types, mode, timeout, use_anycasting, null);
  }

  public RspList callRemoteMethods(String callId, Vector dests, String method_name, Object[] args, Class[] types, int mode, long timeout, boolean use_anycasting,
      RspFilter filter) {
    MethodCall method_call = new MethodCall(method_name, args, types);
    method_call.put("callId", callId);
    return callRemoteMethods(dests, method_call, mode, timeout, use_anycasting, false, filter);
  }

  @Override
  public Object handle(Message req) {
    Call        call = null;
    Object      body = null;
    MethodCall  method_call = null;
    Object      result = null;
    
    if (server_obj == null){
      return null;
    }
    
    if(req == null || req.getLength() == 0) {
      log.severe("message or message buffer is null");
      return null;
    }

    try {
      body=req_marshaller != null? req_marshaller.objectFromByteBuffer(req.getBuffer()) : req.getObject();
    }
    catch(Throwable e) {
      log.severe("exception marshalling object"+e.toString());
      return e;
    }

    if(!(body instanceof MethodCall)) {
      log.severe("message does not contain a MethodCall object");
      return null;
    }

    method_call=(MethodCall)body;
    String callId = (String)method_call.get("callId");
    
    synchronized (calls) {
      if(calls.containsKey(callId)){
        System.out.println("Duplicate call("+callId+") "+ method_call );
        call = calls.get(callId);
      }
      else{
        System.out.println("New call("+callId+") "+ method_call );
        call = new Call(false);
        calls.put(callId, call);
      }
    }
    
    synchronized (call) {
      if(call.valid){
        result = call.result;
      }else{
        result = super.handle(req);
        call.result = result;
        call.valid = true;
      }
    }
    System.out.println("return: "+result);
    return result;
  }
  
  protected class Call{
    public boolean valid;
    public Object result;
    public Call(boolean valid){
      this.valid = valid;
    }
  }
}
